/**
 * Copyright 2008-2009. Chongqing Communications Industry Services Co.,Ltd Information Technology Branch. All rights
 * reserved. <a>http://www.cqcis.com</a>
 */
package com.cqcis.uip.base.framework.handlers.common;


import java.util.concurrent.CountDownLatch;

import org.apache.commons.httpclient.HostConfiguration;
import org.apache.commons.httpclient.HttpConnection;
import org.apache.commons.httpclient.HttpConnectionManager;
import org.apache.commons.httpclient.HttpMethod;
import org.apache.commons.httpclient.SimpleHttpConnectionManager;
import org.apache.commons.httpclient.params.HttpClientParams;
import org.apache.commons.httpclient.params.HttpConnectionParams;
import org.apache.commons.httpclient.protocol.Protocol;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.cqcis.uip.base.framework.FrameworkConstants;
import com.cqcis.uip.base.framework.UIPServiceException;
import com.cqcis.uip.base.framework.common.HttpClient;
import com.cqcis.uip.base.framework.common.InterfaceLog;
import com.cqcis.uip.base.framework.common.Role;
import com.cqcis.uip.base.framework.common.State;
import com.cqcis.uip.base.framework.common.Task;
import com.cqcis.uip.base.framework.core.Endpoint;
import com.cqcis.uip.base.framework.core.RegisteredService;
import com.cqcis.uip.base.framework.core.RegisteredServiceContext;
import com.cqcis.uip.base.framework.core.Service;
import com.cqcis.uip.base.framework.core.ServiceRole;
import com.cqcis.uip.base.framework.core.TaskResultProcessor;
import com.cqcis.uip.base.framework.core.ThreadPoolConfig;
import com.cqcis.uip.base.framework.core.impl.AbstractTaskContext;
import com.cqcis.uip.base.framework.core.impl.HttpClientTaskContextAdaptor;
import com.cqcis.uip.base.framework.core.impl.HttpClientTaskContextAdaptor.HttpClientTaskContextWriter;
import com.cqcis.uip.base.framework.handlers.CommonService;
import com.cqcis.uip.base.framework.handlers.HttpClientService;
import com.cqcis.uip.base.framework.service.InterfaceLogService;
import com.cqcis.uip.base.framework.service.StreamNameUtil;
import com.cqcis.uip.base.framework.service.StreamOutput;
import com.cqcis.uip.base.framework.service.impl.DefaultInterfaceLogService;
import com.cqcis.uip.base.framework.service.impl.HttpClientProtocolSocketFactoryImpl;
import com.cqcis.uip.base.framework.service.impl.HttpConnectionSupport;
import com.cqcis.uip.base.framework.utils.ByteBuffer;
import com.cqcis.uip.base.framework.utils.ConcurrentTaskQueue;
import com.cqcis.uip.base.framework.utils.HttpUtil;
import com.cqcis.uip.base.framework.utils.LogServiceThreadPoolExecutor;
import com.cqcis.uip.base.framework.utils.StringUtil;
import com.cqcis.uip.base.framework.utils.TaskUtil;
import com.cqcis.uip.base.framework.utils.LogServiceThreadPoolExecutor.LogThreadPoolExecutor;

/**
 * HTTP CLIENT客户端调用代理
 * 
 * @author huangmy
 * @date   2008-10-13
 */
@SuppressWarnings("unchecked")
public class AbstractHttpClientHandler implements CommonService, Service, TaskResultProcessor { 
	private static Log log = LogFactory.getLog(AbstractHttpClientHandler.class);
	private static long DEFAULT_CONNECT_TIMEOUT = 60 * 1000; // 默认的连接超时时间
	private static String BLOCKING_QUEUE = "BLOCKING_QUEUE_" + AbstractHttpClientHandler.class.getName();
	private static String PARAM_TASK = "PARAM_TASK_" + AbstractHttpClientHandler.class.getName();
	private static String TASK_INTF_LOG = "TASK_INTF_LOG_" + AbstractHttpClientHandler.class.getName();
	private static String INTF_LOG_SEND_BUFFER = "INTF_LOG_SEND_BUFFER_" + AbstractHttpClientHandler.class.getName();
	private static String INTF_LOG_RECV_BUFFER = "INTF_LOG_RECV_BUFFER_" + AbstractHttpClientHandler.class.getName();
	private LogThreadPoolExecutor messageRecvTpe;
	private RegisteredServiceContext serviceContext;
	private InterfaceLogService interfaceLogService;
	private HttpClientService httpClientService; // 客户端
	private StreamNameUtil streamNameUtil;
    private StreamOutput   streamOutput;
	private HttpClient httpClient; // 客户端
	private String uri; // URI
	private long timeout; //milliseconds
	private String charset = "UTF-8";
	private boolean streamRecord = true; // 是否记录二进制
	
	public boolean processResult(Task task) {
		if (this.httpClientService instanceof TaskResultProcessor) {
			return ((TaskResultProcessor)this.httpClientService).processResult(task);
		}
    	return false;
    }
	
	/**
	 * 异常
	 * 
	 * @param task
	 * @param t
	 */
	public void catchException(Task task, Throwable t) {
		((ConcurrentTaskQueue<Object>)task.getAttribute(BLOCKING_QUEUE)).offer(t);
	}
	
	private HttpClientTaskContextWriter createHttpClientWriter() {
		return new HttpClientTaskContextWriter() {
			@SuppressWarnings("unchecked")
			public void doHttpClientTaskContextWrite(final Task task,
					final HttpMethod method) throws Exception {				
				messageRecvTpe.execute(new Runnable() {
					public void run() {
						try {
							synchronized (task) {
								doCallMessageRecv(task, method);
							}								
						} catch (Throwable t) {
							AbstractTaskContext<?> taskContext = 
								(AbstractTaskContext<?>)task.getTaskContext();
							if (taskContext != null) {
								taskContext.interrupt();
							}
							((ConcurrentTaskQueue<Object>)task.getAttribute(BLOCKING_QUEUE)).offer(t);
						}
					}
				});	
			}
			
			public void afterNormalFinished(Task task) {				
				((ConcurrentTaskQueue<Object>)task.getAttribute(BLOCKING_QUEUE)).offer(Boolean.TRUE);
			}
		};
	}
	            
    public void start(RegisteredServiceContext serviceContext) throws Exception {
    	if (this.httpClientService instanceof Service) {
    		((Service)this.httpClientService).start(serviceContext);
    	}
    	
    	this.serviceContext = serviceContext;
    	
    	if (this.httpClientService instanceof HttpClientTaskContextAdaptor) {
    		ThreadPoolConfig tpConf = serviceContext.getRegisteredService().getDefaultThreadPoolConfig();
    		this.messageRecvTpe = (LogThreadPoolExecutor)LogServiceThreadPoolExecutor.getLogExecutor(tpConf.getCoreSize(), 
    				tpConf.getMaxSize(), tpConf.getKeepAliveTime(), serviceContext,
    				serviceContext.getRegisteredService().getName());
    		
    		HttpClientTaskContextAdaptor httpClientTC = (HttpClientTaskContextAdaptor)this.httpClientService;
    		httpClientTC.setWriter(createHttpClientWriter());
    	}
    	
    	Protocol.unregisterProtocol("http");
    	long connectTimeout = this.timeout/3 + this.timeout%3;
    	connectTimeout = (connectTimeout > DEFAULT_CONNECT_TIMEOUT) ? (DEFAULT_CONNECT_TIMEOUT) : connectTimeout;
		Protocol http = new Protocol("http", new HttpClientProtocolSocketFactoryImpl(connectTimeout), 80);
        Protocol.registerProtocol("http", http);
    	
    	HttpClientParams param = new HttpClientParams();
    	param.setConnectionManagerTimeout(this.timeout); // milliseconds
    	    	
    	this.httpClient = new HttpClient(param, this.uri);
    	this.httpClient.setHttpConnectionManager(createRecordHttpConnectionManager());// 链接管理
    	
    	this.httpClient.getHttpConnectionManager().getParams().setConnectionTimeout((int)this.timeout);
    	this.httpClient.getHttpConnectionManager().getParams().setSoTimeout((int)this.timeout);
    	this.httpClient.getHttpConnectionManager().getParams().setParameter(HttpConnectionParams.CONNECTION_TIMEOUT, Integer.valueOf((int)this.timeout));
    	this.httpClient.getParams().setConnectionManagerTimeout(this.timeout);
    	this.httpClient.getParams().setContentCharset(this.charset);
    }
    
    /**
     * 连接管理
     * 
     * @return
     */
    private HttpConnectionManager createRecordHttpConnectionManager() {
    	return new SimpleHttpConnectionManager() {
    		@Override
    		public HttpConnection getConnection(
    				HostConfiguration hostConfiguration) {
    			HttpConnection connection = super.getConnection(hostConfiguration);
    			if (connection == null) {
    				return null;
    			}
    			
    			return getRecordHttpConnection(hostConfiguration, connection);
    		}
    		
    		@Override
    		public HttpConnection getConnectionWithTimeout(
    				HostConfiguration hostConfiguration, long timeout) {
    			HttpConnection connection = super.getConnectionWithTimeout(hostConfiguration, timeout);
    			if (connection == null) {
    				return null;
    			}
    			
    			return getRecordHttpConnection(hostConfiguration, connection);
    		}
    		
    		@SuppressWarnings("deprecation")
			@Override
    		public HttpConnection getConnection(
    				HostConfiguration hostConfiguration, long timeout) {
    			HttpConnection connection = super.getConnection(hostConfiguration, timeout);
    			if (connection == null) {
    				return null;
    			}
    			
    			return getRecordHttpConnection(hostConfiguration, connection);
    		}
    	};
    }
    
    private HttpConnection getRecordHttpConnection(HostConfiguration hostConfiguration, HttpConnection connection) {
    	Task task = (Task)hostConfiguration.getParams().getParameter(PARAM_TASK);    			
				
    	InterfaceLog intfLog = createInterfaceLog(task);
		task.setAttribute(TASK_INTF_LOG, intfLog);
		
		ByteBuffer sendBuffer = ByteBuffer.allocate(1024).setAutoExpand(true);
		task.setAttribute(INTF_LOG_SEND_BUFFER, sendBuffer);
		
		ByteBuffer recvBuffer = ByteBuffer.allocate(1024).setAutoExpand(true);
		task.setAttribute(INTF_LOG_RECV_BUFFER, recvBuffer);
		
		return new HttpConnectionSupport(connection, sendBuffer, recvBuffer, intfLog);
    }
    
    /**
     * 执行
     * 
     * @param task
     * @param method
     * @throws Exception
     */
    private void doCallMessageRecv(Task task, HttpMethod method) throws Exception {
    	boolean beforeGenerateResult = task.isGenerateResult();
		int beforeGenerateResultCount = task.getGenerateResultCount();
		task.setExecuteDefault(true);
		
		doHttpExecMethod(method, task); // do write and receive
		
		boolean generateResult = TaskUtil.isGenerateResultForCompareWithBefore(task, beforeGenerateResultCount, beforeGenerateResult);
		
		if (task.isExecuteDefault() && generateResult) {
			((ConcurrentTaskQueue<Object>)task.getAttribute(BLOCKING_QUEUE)).offer(Boolean.TRUE);
		}
    }
    
    public void stop(RegisteredServiceContext serviceContext) throws Exception {
    	if (this.httpClientService instanceof Service) {
    		((Service)this.httpClientService).stop(serviceContext);
    	}
    	
    	if (this.messageRecvTpe != null) {
    		this.messageRecvTpe.shutdownWait();
    	}
    	
    	this.serviceContext = null;
    }
    
    private void recordInterfaceLog(InterfaceLog intflog, Exception th) {
    	try {
			if (intflog != null) {
				String recvRecord = intflog.getRecvRecord();
				if (!StringUtil.hasText(recvRecord) && th != null) {
					intflog.setRecvRecord(th.getMessage());
				}
				
				String sendRecord = intflog.getSendRecord();
				if (!StringUtil.hasText(sendRecord) && th != null) {
					intflog.setSendRecord(th.getMessage());
				}
				
				getInterfaceLogService().insert(intflog);
			}
		} catch (Exception e) {
			log.error("Save tin_log Exception:", e);
		}
    }
    
    private InterfaceLog createInterfaceLog(Task task) {
    	return new InterfaceLog(task.getTaskId(), this.serviceContext.getRegisteredService().getId());
    }
    
    /**
     * 获取interfacelog service
     * 
     * @return
     */
    protected InterfaceLogService getInterfaceLogService() {
		if (this.interfaceLogService == null) {
			this.interfaceLogService = new DefaultInterfaceLogService();
		}
		
		return this.interfaceLogService;
	}
        
	@SuppressWarnings("unchecked")
	public void schedule(Task task) throws Exception {	
		boolean blocking = false;		
		HttpMethod method = null;
		
		if (this.httpClientService instanceof HttpClientTaskContextAdaptor) {
			task.setAttribute(BLOCKING_QUEUE, new ConcurrentTaskQueue<Object>());
		}
				
		synchronized (task) {
			try {
				method = this.httpClientService.doMessageSend(this.httpClient, task);
				if (method == null) {
					if (!task.isExecuteDefault()) {
						blocking = true;
					} else {
						throw new NullPointerException(
								"doMessageSend Can't return NULL.");
					}
				}
			} finally {
				CountDownLatch countLatch = (CountDownLatch) task
						.removeAttribute(FrameworkConstants.TASK_BLOCKINNG_PROCESS_SYNC);
				if (countLatch != null) {
					countLatch.countDown();
				}
			}
		}
		
		if (blocking) {			
			Object t = ((ConcurrentTaskQueue<Object>)task.getAttribute(BLOCKING_QUEUE)).poll();// block to wait
			if (t != null && t instanceof Throwable) {
				throw new UIPServiceException((Throwable)t);
			}
			return;
		}
		
		doHttpExecMethod(method, task); // 执行
	}
	
	private void dispatchTaskResultForRemoteDealing(Task task) {
		Task result = task.copy();
		result.setState(State.SUCCESS);
		result.setMessage(FrameworkConstants.TASK_STATE_MSG_REMOTE_DEALNG);
		result.setAttribute(FrameworkConstants.TASK_IGNORE_AFTERPROCESSRESULT, Boolean.TRUE);
		
		Endpoint fromEndpoint = task.route().from();
		RegisteredService srcRS = this.serviceContext.getServiceRepository()
		        .getRegisteredService(fromEndpoint); // 任务源服务
		if (srcRS != null) {
			result.setPriority(result.getPriority() + 1);
			srcRS.processResult(result);
		}
	}
	
	/**
	 * 执行
	 * 
	 * @param method
	 * @param task
	 * @throws Exception
	 */
	private void doHttpExecMethod(HttpMethod method, Task task) throws Exception {
		synchronized (this.httpClient) {
			forHttpExecMethod(method, task);
		}		
	}
	
	/**
	 * 执行
	 */
	private void forHttpExecMethod(HttpMethod method, Task task) throws Exception {		
		this.httpClient.getHostConfiguration().getParams().setParameter(PARAM_TASK, task); 
		
		Exception th = null;
		
		dispatchTaskResultForRemoteDealing(task);
		try {
			this.httpClient.executeMethod(method);
			if (HttpUtil.isOK(method)) {
				this.httpClientService.doMessageReceived(method, task);
			} else {
				throw new UIPServiceException("Method ERROR:" 
						+ method.getStatusCode() + "-"
						+ method.getStatusText());
			}
		} catch (Exception e) {
			th = e;
			throw e;
		} finally {
			try {
			    recordStreamAndLogForTask(task, th);
			} catch (Throwable t) {				
			}
			
			try {
			    method.releaseConnection();
			} catch (Throwable t) {				
			}
		}
	}
	
	private void recordStreamAndLogForTask(Task task, Exception th) throws Exception {
		if (!this.streamRecord) {
			return;
		}
		
		InterfaceLog intfLog = (InterfaceLog)task.getAttribute(TASK_INTF_LOG);
		ByteBuffer sendBuffer = (ByteBuffer)task.getAttribute(INTF_LOG_SEND_BUFFER);			
		ByteBuffer recvBuffer = (ByteBuffer)task.getAttribute(INTF_LOG_RECV_BUFFER);
				
		if (sendBuffer != null) {
			sendBuffer.flip();
		}
		if (recvBuffer != null) {
			recvBuffer.flip();
		}
				
		String filePath = streamNameUtil.getFilePath();		
		String file = null;
		String sendName = streamNameUtil.getSendName();
        if (filePath != null && sendBuffer != null && sendBuffer.hasRemaining()) {
        	file = streamOutput.write(filePath, sendName
					  , sendBuffer.buf());
        	intfLog.setSendRecord(getRecordPathFrom(file));
        }
        
        String nameSuffix = streamNameUtil.getPreFileSuffix();
        
        if (filePath != null && nameSuffix != null && recvBuffer != null && recvBuffer.hasRemaining()) {
			file = streamOutput.write(filePath,
					streamNameUtil.getRecvName(nameSuffix), recvBuffer.buf());
			intfLog.setRecvRecord(getRecordPathFrom(file));
		}
        
		recordInterfaceLog(intfLog, th);
	}
	
	/**
	 * 二进制日志路径
	 * 
	 * @param file
	 * @return
	 */
	protected String getRecordPathFrom(String file) {
		if (file == null) {
			return null;
		}
		int splitBegin = file.indexOf(FrameworkConstants.STREAM_LOG_PATH_STOKEN);
		if (splitBegin >= 0) {
			splitBegin = splitBegin + FrameworkConstants.STREAM_LOG_PATH_STOKEN.length() + 1;
		} else {
			splitBegin = 0;
		}
		
		return file.substring(splitBegin);
	}
    	
	@ServiceRole(Role.HTTP_CLIENT)
	public void setWebserviceCaller(HttpClientService httpClientService) {
		this.httpClientService = httpClientService;
	}

	public void setUri(String uri) {
		this.uri = uri;
	}

	public void setTimeout(long timeout) {
		this.timeout = timeout;
	}

	public void setCharset(String charset) {
		this.charset = charset;
	}
	
	public void setInterfaceLogService(InterfaceLogService interfaceLogService) {
		this.interfaceLogService = interfaceLogService;
	}
	
	public void setStreamNameUtil(StreamNameUtil streamNameUtil) {
		this.streamNameUtil = streamNameUtil;
	}

	public void setStreamOutput(StreamOutput streamOutput) {
		this.streamOutput = streamOutput;
	}

	public void setStreamRecord(boolean streamRecord) {
		this.streamRecord = streamRecord;
	}
}
